Swarm 框架简易入门指南

#AI

什么是 Swarm?

Swarm 是一个简单但功能强大的 Python 框架,它让多个智能助手(称为 Agent)能够协同工作并相互交流。想象一下,这就像一群小蜜蜂一起完成不同任务,每只蜜蜂都有特定技能,它们可以把工作传递给其他更擅长该任务的蜜蜂。

🔔 重要提示:Swarm 现已被 OpenAI Agents SDK 替代,后者是更新更完善的版本。本指南帮助你理解 Swarm 的基本概念,这些概念与新的 Agents SDK 类似。

核心概念(基础构建块)

1. Agent(智能助手)

Agent 就像一个聪明的小助手,它知道如何完成特定任务。每个 Agent 都有:

  • 名字(例如"天气助手"或"数学助手")
  • 指令(告诉 Agent 应该做什么和如何交谈)
  • 函数(Agent 可以使用的特殊工具)

案例1:创建一个简单的助手

from swarm import Swarm, Agent
 
simple_agent = Agent(
    name="简单助手",
    instructions="你是一个乐于助人的助手,总是用友善的语气回答问题。",
)

案例2:创建一个天气助手

weather_agent = Agent(
    name="天气助手",
    instructions="你是专门提供天气信息的助手,会告诉用户任何地点的天气。",
    functions=[get_weather, send_email]
)

案例3:创建一个翻译助手

translator_agent = Agent(
    name="翻译助手",
    instructions="你是语言翻译专家,可以将任何语言翻译成用户想要的语言。",
    functions=[translate_text]
)

2. 函数(Functions)

函数就像是 Agent 可以使用的特殊工具。它们让 Agent 能够执行实际操作,比如:

  • 查询天气信息
  • 发送电子邮件
  • 翻译文本

案例1:天气查询函数

def get_weather(location):
    """获取指定地点的天气。地点必须是一个城市名。"""
    # 这里会有获取天气的实际代码
    return f"{location} 今天天气晴朗,气温25°C。"

案例2:发送电子邮件函数

def send_email(recipient, subject, body):
    """发送一封电子邮件。"""
    print(f"正在发送邮件...")
    print(f"收件人: {recipient}")
    print(f"主题: {subject}")
    print(f"内容: {body}")
    return "邮件已发送成功!"

案例3:翻译文本函数

def translate_text(text, target_language):
    """将文本翻译成目标语言。"""
    # 这里会有翻译代码
    return f"已将文本翻译成{target_language}: [翻译后的文本]"

3. 交接(Handoffs)

交接是指一个 Agent 将对话传递给另一个更合适的 Agent。这就像一个工作人员把你转接给更专业的同事一样。

案例1:将用户转至西班牙语助手

def transfer_to_spanish_agent():
    """将用户转给西班牙语助手处理。"""
    return spanish_agent

案例2:将用户转至技术支持

def transfer_to_tech_support():
    """如果用户有技术问题,将其转给技术支持团队。"""
    return tech_support_agent

案例3:根据问题类型转接

def transfer_based_on_topic(topic):
    """根据话题将用户转给相应的专家。"""
    if topic == "退款":
        return refund_agent
    elif topic == "订单":
        return order_agent
    else:
        return general_agent

4. 上下文变量(Context Variables)

上下文变量是在对话期间可以在不同 Agent 之间共享的信息片段。它们让 Agent 能够记住用户的信息和之前的对话内容。

案例1:记住用户名称

# 创建包含用户信息的上下文变量
context_variables = {"user_name": "小明", "user_id": 123}
 
# 使用这些变量运行对话
response = client.run(
    agent=greeter_agent,
    messages=[{"role": "user", "content": "你好!"}],
    context_variables=context_variables
)
# 助手会回复:"你好,小明!很高兴见到你。"

案例2:在指令中使用上下文变量

def instructions(context_variables):
    name = context_variables.get("user_name", "用户")
    return f"你是一个乐于助人的助手。请用用户的名字({name})来打招呼。"
 
personalized_agent = Agent(
    name="个性化助手",
    instructions=instructions
)

案例3:在函数中使用上下文变量

def print_account_details(context_variables):
    user_id = context_variables.get("user_id", "未知")
    name = context_variables.get("user_name", "未知用户")
    print(f"账户详情: 用户名={name}, ID={user_id}")
    return "账户信息已显示"

Swarm 工作流程(图解)

┌─────────────────┐     ┌─────────────────┐     ┌─────────────────┐
│                 │     │                 │     │                 │
│     用户输入     │────▶│    Swarm 客户端  │────▶│    初始 Agent   │
│                 │     │                 │     │                 │
└─────────────────┘     └─────────────────┘     └────────┬────────┘


┌─────────────────┐     ┌─────────────────┐     ┌─────────────────┐
│                 │     │                 │     │                 │
│   返回最终回复   │◀────│    处理结果      │◀────│  Agent 处理请求  │
│                 │     │                 │     │                 │
└─────────────────┘     └─────────────────┘     └────────┬────────┘
                                 ▲                       │
                                 │                       ▼
                                 │               ┌─────────────────┐
                                 │               │                 │
                                 │               │ 需要使用函数吗?  │
                                 │               │                 │
                                 │               └────────┬────────┘
                                 │                        │
                                 │                        ▼
                          ┌──────┴───────┐       ┌─────────────────┐
                          │              │ 是    │                 │
                          │  需要交接吗?  │◀─────│   执行函数调用   │
                          │              │       │                 │
                          └──────┬───────┘       └─────────────────┘
                                 │ 否


                        ┌─────────────────┐
                        │                 │
                        │  Agent 提供回复  │
                        │                 │
                        └─────────────────┘

Swarm 代码结构

  1. core.py: 主要引擎,使所有功能协同工作
  2. types.py: 定义 Agent、Response 和 Result 的结构
  3. util.py: 框架使用的辅助工具
  4. repl: 让你可以与 Agent 聊天的工具

如何使用 Swarm(简单步骤)

步骤 1:设置环境

首先安装 Swarm(需要 Python 3.10+):

pip install git+https://github.com/openai/swarm.git

步骤 2:创建一个简单的 Agent

from swarm import Swarm, Agent
 
# 创建 Swarm 客户端
client = Swarm()
 
# 创建一个简单的助手 Agent
helper = Agent(
    name="我的助手",
    instructions="你是一个友好的助手,总是用热情的语气回答问题。"
)

步骤 3:与 Agent 交谈

# 发送消息并获取回复
response = client.run(
    agent=helper,
    messages=[{"role": "user", "content": "你好!"}]
)
 
# 查看 Agent 的回复
print(response.messages[-1]["content"])

进阶功能演示

1. 使用函数功能

案例:创建一个可以查询天气的助手

from swarm import Swarm, Agent
 
# 创建 Swarm 客户端
client = Swarm()
 
# 定义天气查询函数
def get_weather(location, time="现在"):
    """获取指定地点的天气。地点必须是一个城市名。"""
    return f"{location}{time}的天气晴朗,气温25°C。"
 
# 创建一个带有函数的助手
weather_assistant = Agent(
    name="天气助手",
    instructions="你是一个天气助手,可以告诉用户任何地点的天气情况。",
    functions=[get_weather]
)
 
# 与助手交谈
response = client.run(
    agent=weather_assistant,
    messages=[{"role": "user", "content": "北京今天天气怎么样?"}]
)
 
# 查看回复
print(response.messages[-1]["content"])

2. 使用 Agent 交接功能

案例:创建可以相互交接的多语言助手

from swarm import Swarm, Agent
 
# 创建 Swarm 客户端
client = Swarm()
 
# 创建中文助手
chinese_agent = Agent(
    name="中文助手",
    instructions="你只说中文,用友好的语气回答问题。"
)
 
# 创建英文助手
english_agent = Agent(
    name="English Assistant",
    instructions="You only speak English. Be friendly and helpful."
)
 
# 创建交接函数
def transfer_to_chinese():
    """将用户转给中文助手"""
    return chinese_agent
 
def transfer_to_english():
    """Transfer to English assistant"""
    return english_agent
 
# 添加交接函数到助手
chinese_agent.functions.append(transfer_to_english)
english_agent.functions.append(transfer_to_chinese)
 
# 开始与中文助手交谈
response = client.run(
    agent=chinese_agent,
    messages=[{"role": "user", "content": "Can you speak English?"}]
)
 
# 助手会检测到英文并转接到英文助手
print(response.messages[-1]["content"])

3. 使用上下文变量

案例:创建能记住用户偏好的助手

from swarm import Swarm, Agent, Result
 
# 创建 Swarm 客户端
client = Swarm()
 
# 定义保存偏好的函数
def save_preference(preference_type, value):
    """保存用户偏好"""
    return Result(
        value=f"已保存您的{preference_type}偏好: {value}",
        context_variables={preference_type: value}
    )
 
# 定义获取偏好的函数
def get_preference(context_variables, preference_type):
    """获取用户之前保存的偏好"""
    value = context_variables.get(preference_type, "未设置")
    return f"您的{preference_type}偏好是: {value}"
 
# 创建带有这些函数的助手
preference_agent = Agent(
    name="偏好助手",
    instructions="你是一个可以记住用户偏好的助手。",
    functions=[save_preference, get_preference]
)
 
# 用户设置偏好
response1 = client.run(
    agent=preference_agent,
    messages=[{"role": "user", "content": "请记住我喜欢蓝色"}],
    context_variables={}
)
 
# 获取更新后的上下文变量
updated_context = response1.context_variables
 
# 用户询问之前设置的偏好
response2 = client.run(
    agent=preference_agent,
    messages=[{"role": "user", "content": "我之前说我喜欢什么颜色?"}],
    context_variables=updated_context
)
 
# 查看回复
print(response2.messages[-1]["content"])

实用案例详解

1. 客户服务系统

from swarm import Swarm, Agent
 
# 创建不同类型的客服助手
triage_agent = Agent(
    name="分类助手",
    instructions="你的工作是确定用户的问题类型,然后转接到相应的专业助手。"
)
 
refund_agent = Agent(
    name="退款助手",
    instructions="你是退款专家,可以处理所有与退款相关的问题。"
)
 
order_agent = Agent(
    name="订单助手",
    instructions="你是订单专家,可以处理所有与订单相关的问题。"
)
 
# 创建转接函数
def transfer_to_refund():
    """转接到退款助手"""
    return refund_agent
 
def transfer_to_order():
    """转接到订单助手"""
    return order_agent
 
# 添加转接函数到分类助手
triage_agent.functions = [transfer_to_refund, transfer_to_order]
 
# 使用分类助手开始对话
client = Swarm()
response = client.run(
    agent=triage_agent,
    messages=[{"role": "user", "content": "我想申请退款,我的订单号是123456"}]
)
 
# 分类助手会将用户转接到退款助手
print(f"当前处理的助手: {response.agent.name}")
print(f"助手回复: {response.messages[-1]['content']}")

2. 天气提醒系统

from swarm import Swarm, Agent, Result
 
# 天气检查函数
def check_weather(location):
    """检查指定地点的天气"""
    # 假设这里有真实的天气API调用
    return "晴天,温度25°C"
 
# 发送提醒函数
def send_reminder(context_variables, message):
    """向用户发送提醒"""
    user_name = context_variables.get("user_name", "用户")
    print(f"发送提醒给 {user_name}: {message}")
    return f"提醒已发送"
 
# 保存用户位置函数
def save_location(location):
    """保存用户的位置偏好"""
    return Result(
        value=f"已保存您的位置: {location}",
        context_variables={"user_location": location}
    )
 
# 创建天气助手
weather_agent = Agent(
    name="天气助手",
    instructions="你是一个天气助手,可以检查天气并发送提醒。",
    functions=[check_weather, send_reminder, save_location]
)
 
# 运行对话
client = Swarm()
response = client.run(
    agent=weather_agent,
    messages=[
        {"role": "user", "content": "我住在上海,请记住这一点"},
        {"role": "assistant", "content": "好的,我已记住您住在上海。"},
        {"role": "user", "content": "现在天气怎么样?如果是好天气,请提醒我出去散步"}
    ],
    context_variables={"user_name": "张先生"}
)
 
# 查看结果
print(response.messages[-1]["content"])

3. 多语言翻译链

from swarm import Swarm, Agent
 
# 创建翻译函数
def translate_to_english(text):
    """将文本翻译成英文"""
    # 假设这里有真实的翻译API
    return f"英文翻译: {text} -> [英文翻译结果]"
 
def translate_to_chinese(text):
    """将文本翻译成中文"""
    return f"中文翻译: {text} -> [中文翻译结果]"
 
def translate_to_spanish(text):
    """将文本翻译成西班牙语"""
    return f"西班牙语翻译: {text} -> [西班牙语翻译结果]"
 
# 创建识别语言的函数
def detect_language(text):
    """识别文本的语言"""
    # 简单示例逻辑
    if "你好" in text or "中文" in text:
        return "中文"
    elif "hola" in text or "español" in text:
        return "西班牙语"
    else:
        return "英文"
 
# 创建转接函数
def transfer_to_appropriate_translator(text):
    """根据检测到的语言转接到适当的翻译助手"""
    language = detect_language(text.lower())
    if language == "中文":
        return chinese_translator
    elif language == "西班牙语":
        return spanish_translator
    else:
        return english_translator
 
# 创建各种翻译助手
language_detector = Agent(
    name="语言识别助手",
    instructions="你帮助识别用户使用的语言,并转接到相应的翻译助手。",
    functions=[transfer_to_appropriate_translator]
)
 
english_translator = Agent(
    name="英文翻译助手",
    instructions="你是英文翻译专家。",
    functions=[translate_to_english]
)
 
chinese_translator = Agent(
    name="中文翻译助手",
    instructions="你是中文翻译专家。",
    functions=[translate_to_chinese]
)
 
spanish_translator = Agent(
    name="西班牙语翻译助手",
    instructions="你是西班牙语翻译专家。",
    functions=[translate_to_spanish]
)
 
# 启动对话
client = Swarm()
response = client.run(
    agent=language_detector,
    messages=[{"role": "user", "content": "Hola, me gustaría traducir esto al chino."}]
)
 
# 查看结果
print(f"当前助手: {response.agent.name}")
print(response.messages[-1]["content"])

测试你的 Swarm 系统

Swarm 提供了测试工具,帮助你确保你的 Agent 正常工作:

from swarm import Swarm, Agent
from tests.mock_client import MockOpenAIClient, create_mock_response
 
# 创建模拟客户端
mock_client = MockOpenAIClient()
mock_client.set_response(
    create_mock_response({"role": "assistant", "content": "我是测试回复"})
)
 
# 使用模拟客户端创建 Swarm
client = Swarm(client=mock_client)
 
# 创建测试 Agent
test_agent = Agent(name="测试助手")
 
# 运行测试
response = client.run(
    agent=test_agent,
    messages=[{"role": "user", "content": "测试消息"}]
)
 
# 验证结果
assert response.messages[-1]["content"] == "我是测试回复"
print("测试通过!")

实用开发技巧

  1. 从简单开始:先创建一个基本的 Agent,确保它能正常回答问题
  2. 逐步添加函数:一次添加一个函数,并测试每个函数是否正常工作
  3. 使用有意义的名称:给 Agent 和函数起有描述性的名称
  4. 清晰的指令:Agent 的指令应该非常明确,告诉它应该做什么和不应该做什么
  5. 使用上下文变量:利用上下文变量在多个交互中保持状态
  6. 测试每个路径:确保所有可能的交接路径都能正常工作
  7. 使用 run_demo_loop 进行快速测试
    from swarm.repl import run_demo_loop
    run_demo_loop(my_agent, stream=True)

总结:Swarm 的优势

  • 轻量级:简单,不需要复杂设置
  • 灵活:可以轻松更改 Agent 的行为
  • 可组合:Agent 可以以多种方式协同工作
  • 无状态:不需要在调用之间保存复杂的信息

Swarm 框架让你可以轻松创建由多个智能助手组成的系统,它们可以协同工作、相互交流,并使用特殊功能来帮助用户。无论你是想创建一个简单的问答机器人,还是一个复杂的多助手系统,Swarm 都提供了简单而强大的工具来实现它!